package com.casic.quality.generator.service.impl;

import com.casic.quality.generator.dto.RuleInfo;
import com.casic.quality.generator.service.GeneratorService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.*;

/**
 * class description
 *
 * @author 朱宗刚
 * @version 1.0.0
 * @date 2020/9/23
 */

@Service
public class GeneratorServiceImpl implements GeneratorService {
    /**
     * 根据规则生成flink应用程序包
     *
     * @param ruleInfo 规则描述
     * @return 是否成功
     */
    @Override
    public Object ruleGenerator(RuleInfo ruleInfo) {
        try (FileWriter writer = new FileWriter(base + File.separator + path + File.separator + "AutoGenerator.scala")) {
            writer.write("package " + path.replaceAll("\\\\", ".") + "\n\n");
            writer.write("import org.apache.flink.api.scala._\n\n");
            writer.write("object MyAggAuto extends App {\n");
            writer.write("\tval env = ExecutionEnvironment.getExecutionEnvironment\n");
            writer.write("\tval dataset1 = env.fromCollection(List((\"a\", 1, 1.1), (\"b\", 2, 2.2), (\"c\", 3, 3.3), (\"b\", 4, 4.4), (\"c\", 5, 5.5)))\n");
            writer.write("\tdataset1.sum(2).max(1).print()\n");
            writer.write("}\n");
            writer.flush();

            Process process = Runtime.getRuntime().exec(pkgPath);
            process.waitFor();
            BufferedReader br = new BufferedReader(new InputStreamReader(process.getInputStream()));
            String line;
            while ((line=br.readLine())!=null) {
                logger.debug(line);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Value("${casic.flink.pkg}")
    private String pkgPath;

    @Value("${casic.flink.path}")
    private String path;

    @Value("${casic.flink.base}")
    private String base;

    private static final Logger logger = LoggerFactory.getLogger(GeneratorServiceImpl.class);
}
